#define _POSIX_C_SOURCE 200112L
#define _DARWIN_C_SOURCE
#include "query_planner.h"
#include "../stats/statistics_collector.h"
#include "../config/config_manager.h"
#include "../memory/memory_manager.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <math.h>
#include <inttypes.h>

// Constants for cost estimation
#define CPU_COST_FACTOR      1.0
#define IO_COST_FACTOR       4.0
#define MEMORY_COST_FACTOR   0.1
#define SELECTIVITY_DEFAULT  0.1
#define ROWS_PER_PAGE        100

// Default optimization parameters
#define DEFAULT_MAX_ITERATIONS     1000
#define DEFAULT_COST_THRESHOLD     1e6
#define DEFAULT_MEMO_TABLE_SIZE    1024

// Helper functions
static uint32_t hash_group_id(uint32_t group_id, uint32_t table_size);
static uint64_t get_current_time_us(void);
static plan_node_t* create_plan_node(plan_node_type_t type);
static void init_default_rules(optimizer_context_t* ctx);
static plan_node_t* planner_create_sort_plan_from_order_by(order_by_item_t* order_by_list, plan_node_t* child);

// Plan creation functions - using statement union members instead of separate types
static plan_node_t* planner_create_select_plan(query_planner_t* planner, statement_t* stmt);
static plan_node_t* planner_create_insert_plan(query_planner_t* planner, statement_t* stmt);
static plan_node_t* planner_create_update_plan(query_planner_t* planner, statement_t* stmt);
static plan_node_t* planner_create_delete_plan(query_planner_t* planner, statement_t* stmt);

// Utility function implementations
static uint32_t hash_group_id(uint32_t group_id, uint32_t table_size) {
    return group_id % table_size;
}

static uint64_t get_current_time_us(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (uint64_t)ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
}

static plan_node_t* create_plan_node(plan_node_type_t type) {
    plan_node_t* node = calloc(1, sizeof(plan_node_t));
    if (!node) return NULL;
    
    static uint32_t next_node_id = 1;
    node->type = type;
    node->node_id = next_node_id++;
    node->optimization_level = 0;
    
    return node;
}

// Query planner API implementation
query_planner_t* planner_create(void) {
    query_planner_t* planner = calloc(1, sizeof(query_planner_t));
    if (!planner) return NULL;
    
    planner->optimizer = optimizer_create_context();
    if (!planner->optimizer) {
        free(planner);
        return NULL;
    }
    
    planner->enable_optimization = true;
    planner->optimization_level = 1; // Basic optimization
    
    return planner;
}

void planner_destroy(query_planner_t* planner) {
    if (!planner) return;
    
    if (planner->optimizer) {
        optimizer_destroy_context(planner->optimizer);
    }
    
    if (planner->current_plan) {
        plan_destroy(planner->current_plan);
    }
    
    free(planner);
}

// Optimizer context functions
optimizer_context_t* optimizer_create_context(void) {
    optimizer_context_t* ctx = calloc(1, sizeof(optimizer_context_t));
    if (!ctx) return NULL;
    
    // Initialize configuration
    ctx->max_iterations = DEFAULT_MAX_ITERATIONS;
    ctx->cost_threshold = DEFAULT_COST_THRESHOLD;
    ctx->enable_join_reordering = true;
    ctx->enable_predicate_pushdown = true;
    ctx->enable_index_selection = true;
    
    // Initialize memo table
    ctx->memo_table_size = DEFAULT_MEMO_TABLE_SIZE;
    ctx->memo_table = calloc(ctx->memo_table_size, sizeof(memo_group_t*));
    if (!ctx->memo_table) {
        free(ctx);
        return NULL;
    }
    
    ctx->next_group_id = 1;
    
    // Initialize integrated systems
    ctx->stats_collector = stats_collector_create();
    ctx->config_manager = config_manager_create("gudb_optimizer.conf");
    
    // Initialize cost model parameters with defaults
    ctx->cpu_cost_factor = CPU_COST_FACTOR;
    ctx->io_cost_factor = IO_COST_FACTOR;
    ctx->memory_cost_factor = MEMORY_COST_FACTOR;
    ctx->index_selectivity_boost = 0.5;
    ctx->hash_join_threshold = 1000;
    
    // Load configuration parameters if config manager is available
    if (ctx->config_manager) {
        optimizer_load_config_params(ctx);
    }
    
    // Initialize built-in rules
    init_default_rules(ctx);
    
    return ctx;
}

void optimizer_destroy_context(optimizer_context_t* ctx) {
    if (!ctx) return;
    
    // Free integrated systems
    if (ctx->stats_collector) {
        stats_collector_destroy(ctx->stats_collector);
    }
    if (ctx->config_manager) {
        config_manager_destroy(ctx->config_manager);
    }
    
    // Free memo table
    if (ctx->memo_table) {
        for (uint32_t i = 0; i < ctx->memo_table_size; i++) {
            memo_group_t* group = ctx->memo_table[i];
            while (group) {
                memo_group_t* next = group->next;
                // Free expressions in group
                plan_node_t* expr = group->expressions;
                while (expr) {
                    plan_node_t* next_expr = expr->next;
                    plan_destroy(expr);
                    expr = next_expr;
                }
                free(group);
                group = next;
            }
        }
        free(ctx->memo_table);
    }
    
    // Free rules
    if (ctx->transformation_rules) {
        free(ctx->transformation_rules);
    }
    if (ctx->implementation_rules) {
        free(ctx->implementation_rules);
    }
    
    // Free table statistics
    if (ctx->table_stats) {
        free(ctx->table_stats);
    }
    
    free(ctx);
}

// Main planning functions
plan_node_t* planner_create_plan(query_planner_t* planner, statement_t* stmt) {
    if (!planner || !stmt) return NULL;
    
    uint64_t start_time = get_current_time_us();
    planner->current_stmt = stmt;
    
    plan_node_t* plan = NULL;
    
    switch (stmt->type) {
        case STMT_SELECT:
            plan = planner_create_select_plan(planner, stmt);
            break;
            
        case STMT_INSERT:
            plan = planner_create_insert_plan(planner, stmt);
            break;
            
        case STMT_UPDATE:
            plan = planner_create_update_plan(planner, stmt);
            break;
            
        case STMT_DELETE:
            plan = planner_create_delete_plan(planner, stmt);
            break;
            
        default:
            return NULL;
    }
    
    if (plan && planner->enable_optimization) {
        plan = planner_optimize_plan(planner, plan);
    }
    
    planner->optimization_time_us += get_current_time_us() - start_time;
    planner->plans_generated++;
    planner->current_plan = plan;
    
    return plan;
}

plan_node_t* planner_optimize_plan(query_planner_t* planner, plan_node_t* plan) {
    if (!planner || !plan || !planner->optimizer) return plan;
    
    optimizer_context_t* ctx = planner->optimizer;
    
    // Create initial memo group
    memo_group_t* root_group = memo_create_group(ctx);
    if (!root_group) return plan;
    
    // Add initial plan to memo
    if (memo_add_expression(ctx, root_group, plan) != 0) {
        return plan;
    }
    
    // Run Cascades optimization
    if (cascades_optimize_group(ctx, root_group) == 0) {
        // Return the best plan found
        if (root_group->best_plan) {
            plan_destroy(plan); // Free original plan
            return plan_clone(root_group->best_plan);
        }
    }
    
    return plan; // Return original plan if optimization failed
}

// Plan creation helper functions
static plan_node_t* planner_create_select_plan(query_planner_t* planner, statement_t* stmt) {
    if (!planner || !stmt || stmt->type != STMT_SELECT) return NULL;
    
    // Simplified SELECT plan creation
    plan_node_t* plan = NULL;
    
    // Create base scan plan
    if (stmt->data.select.from_list) {
        plan = planner_create_scan_plan(stmt->data.select.from_list->table_name, NULL);
    }
    
    // Add WHERE clause as filter
    if (plan && stmt->data.select.where_clause) {
        plan = planner_create_filter_plan(stmt->data.select.where_clause, plan);
    }
    
    // Add projection for SELECT list
    if (plan && stmt->data.select.select_list) {
        // For now, create a simple projection
        expression_t* exprs[] = {stmt->data.select.select_list};
        char* names[] = {"result"};
        plan = planner_create_project_plan(exprs, names, 1, plan);
    }
    
    // Add ORDER BY clause as sort plan
    if (plan && stmt->data.select.order_by) {
        plan = planner_create_sort_plan_from_order_by(stmt->data.select.order_by, plan);
    }
    
    // Add LIMIT clause
    if (plan && stmt->data.select.limit > 0) {
        plan = planner_create_limit_plan(stmt->data.select.limit, stmt->data.select.offset, plan);
    }
    
    return plan;
}

// Basic plan creation functions
plan_node_t* planner_create_scan_plan(const char* table_name, expression_t* filter) {
    plan_node_t* node = create_plan_node(PLAN_SCAN);
    if (!node) return NULL;
    
    node->data.scan.table_name = strdup(table_name ? table_name : "unknown");
    node->data.scan.filter = filter;
    
    // Set default statistics
    node->data.scan.stats.row_count = 1000;
    node->data.scan.stats.page_count = 10;
    node->data.scan.stats.selectivity = 1.0;
    node->data.scan.stats.avg_row_size = 100.0;
    node->data.scan.stats.has_index = false;
    
    return node;
}

plan_node_t* planner_create_index_scan_plan(const char* table_name, const char* index_name, 
                                           expression_t* condition) {
    plan_node_t* node = create_plan_node(PLAN_INDEX_SCAN);
    if (!node) return NULL;
    
    node->data.index_scan.table_name = strdup(table_name ? table_name : "unknown");
    node->data.index_scan.index_name = strdup(index_name ? index_name : "unknown");
    node->data.index_scan.index_condition = condition;
    node->data.index_scan.is_unique = false;
    
    // Set default statistics (index scans are typically more selective)
    node->data.index_scan.stats.row_count = 100;
    node->data.index_scan.stats.page_count = 2;
    node->data.index_scan.stats.selectivity = 0.1;
    node->data.index_scan.stats.avg_row_size = 100.0;
    node->data.index_scan.stats.has_index = true;
    
    return node;
}

plan_node_t* planner_create_filter_plan(expression_t* condition, plan_node_t* child) {
    plan_node_t* node = create_plan_node(PLAN_FILTER);
    if (!node) return NULL;
    
    node->data.filter.condition = condition;
    node->data.filter.child = child;
    
    // Add child to children array
    node->children = malloc(sizeof(plan_node_t*));
    if (node->children) {
        node->children[0] = child;
        node->child_count = 1;
    }
    
    return node;
}

plan_node_t* planner_create_project_plan(expression_t** expressions, char** column_names, 
                                        uint32_t expr_count, plan_node_t* child) {
    plan_node_t* node = create_plan_node(PLAN_PROJECT);
    if (!node) return NULL;
    
    if (expr_count > 0) {
        node->data.project.expressions = malloc(expr_count * sizeof(expression_t*));
        node->data.project.column_names = malloc(expr_count * sizeof(char*));
        
        if (node->data.project.expressions && node->data.project.column_names) {
            for (uint32_t i = 0; i < expr_count; i++) {
                node->data.project.expressions[i] = expressions[i];
                node->data.project.column_names[i] = strdup(column_names[i] ? column_names[i] : "col");
            }
            node->data.project.expr_count = expr_count;
        }
    }
    
    node->data.project.child = child;
    
    // Add child to children array
    node->children = malloc(sizeof(plan_node_t*));
    if (node->children) {
        node->children[0] = child;
        node->child_count = 1;
    }
    
    return node;
}

plan_node_t* planner_create_join_plan(join_type_t join_type, expression_t* condition,
                                     plan_node_t* left, plan_node_t* right) {
    plan_node_t* node = create_plan_node(PLAN_JOIN);
    if (!node) return NULL;
    
    node->data.join.join_type = join_type;
    node->data.join.condition = condition;
    node->data.join.left = left;
    node->data.join.right = right;
    
    // Add children to children array
    node->children = malloc(2 * sizeof(plan_node_t*));
    if (node->children) {
        node->children[0] = left;
        node->children[1] = right;
        node->child_count = 2;
    }
    
    return node;
}

plan_node_t* planner_create_sort_plan(expression_t** sort_keys, bool* ascending,
                                     uint32_t key_count, plan_node_t* child) {
    plan_node_t* node = create_plan_node(PLAN_SORT);
    if (!node) return NULL;
    
    if (key_count > 0) {
        node->data.sort.sort_keys = malloc(key_count * sizeof(expression_t*));
        node->data.sort.ascending = malloc(key_count * sizeof(bool));
        
        if (node->data.sort.sort_keys && node->data.sort.ascending) {
            for (uint32_t i = 0; i < key_count; i++) {
                node->data.sort.sort_keys[i] = sort_keys[i];
                node->data.sort.ascending[i] = ascending ? ascending[i] : true;
            }
            node->data.sort.key_count = key_count;
        }
    }
    
    node->data.sort.child = child;
    
    // Add child to children array
    node->children = malloc(sizeof(plan_node_t*));
    if (node->children) {
        node->children[0] = child;
        node->child_count = 1;
    }
    
    return node;
}

plan_node_t* planner_create_limit_plan(uint64_t limit, uint64_t offset, plan_node_t* child) {
    plan_node_t* node = create_plan_node(PLAN_LIMIT);
    if (!node) return NULL;
    
    node->data.limit.limit = limit;
    node->data.limit.offset = offset;
    node->data.limit.child = child;
    
    // Add child to children array
    node->children = malloc(sizeof(plan_node_t*));
    if (node->children) {
        node->children[0] = child;
        node->child_count = 1;
    }
    
    return node;
}

// Helper function to convert order_by_item_t to sort plan
static plan_node_t* planner_create_sort_plan_from_order_by(order_by_item_t* order_by_list, plan_node_t* child) {
    if (!order_by_list || !child) {
        return child; // No sorting needed
    }
    
    // Count the number of ORDER BY items
    uint32_t key_count = 0;
    order_by_item_t* current = order_by_list;
    while (current) {
        key_count++;
        current = current->next;
    }
    
    if (key_count == 0) {
        return child; // No sorting needed
    }
    
    // Allocate arrays for sort keys and directions
    expression_t** sort_keys = malloc(key_count * sizeof(expression_t*));
    bool* ascending = malloc(key_count * sizeof(bool));
    
    if (!sort_keys || !ascending) {
        free(sort_keys);
        free(ascending);
        return child; // Return original plan on memory error
    }
    
    // Convert order_by_item_t list to arrays
    current = order_by_list;
    for (uint32_t i = 0; i < key_count; i++) {
        if (current) {
            sort_keys[i] = current->expression;
            ascending[i] = (current->direction == SORT_ASC);
            current = current->next;
        }
    }
    
    // Create the sort plan using the existing function
    plan_node_t* sort_plan = planner_create_sort_plan(sort_keys, ascending, key_count, child);
    
    // Clean up temporary arrays (the expressions are referenced, not copied)
    free(sort_keys);
    free(ascending);
    
    return sort_plan;
}

// Memo management functions
memo_group_t* memo_create_group(optimizer_context_t* ctx) {
    if (!ctx) return NULL;
    
    memo_group_t* group = calloc(1, sizeof(memo_group_t));
    if (!group) return NULL;
    
    group->group_id = ctx->next_group_id++;
    group->best_cost.total_cost = INFINITY;
    
    // Add to hash table
    uint32_t hash = hash_group_id(group->group_id, ctx->memo_table_size);
    group->next = ctx->memo_table[hash];
    ctx->memo_table[hash] = group;
    ctx->memo_group_count++;
    
    return group;
}

int memo_add_expression(optimizer_context_t* ctx, memo_group_t* group, plan_node_t* expr) {
    if (!ctx || !group || !expr) return -1;
    
    // Add expression to group's list
    expr->group = group;
    expr->next = group->expressions;
    group->expressions = expr;
    
    return 0;
}

memo_group_t* memo_find_group(optimizer_context_t* ctx, uint32_t group_id) {
    if (!ctx || !ctx->memo_table) return NULL;
    
    uint32_t hash = hash_group_id(group_id, ctx->memo_table_size);
    memo_group_t* group = ctx->memo_table[hash];
    
    while (group) {
        if (group->group_id == group_id) {
            return group;
        }
        group = group->next;
    }
    
    return NULL;
}

// Cost estimation functions
plan_cost_t estimate_plan_cost(plan_node_t* plan, optimizer_context_t* ctx) {
    plan_cost_t cost = {0};
    
    if (!plan) return cost;
    
    switch (plan->type) {
        case PLAN_SCAN: {
            uint64_t rows = plan->data.scan.stats.row_count;
            cost.io_cost = plan->data.scan.stats.page_count * IO_COST_FACTOR;
            cost.cpu_cost = rows * CPU_COST_FACTOR;
            cost.cardinality = rows;
            break;
        }
        
        case PLAN_INDEX_SCAN: {
            uint64_t rows = plan->data.index_scan.stats.row_count;
            cost.io_cost = plan->data.index_scan.stats.page_count * IO_COST_FACTOR;
            cost.cpu_cost = rows * CPU_COST_FACTOR * 0.5; // Index scans are more efficient
            cost.cardinality = rows;
            break;
        }
        
        case PLAN_FILTER: {
            if (plan->data.filter.child) {
                plan_cost_t child_cost = estimate_plan_cost(plan->data.filter.child, ctx);
                cost.io_cost = child_cost.io_cost;
                cost.cpu_cost = child_cost.cpu_cost + child_cost.cardinality * CPU_COST_FACTOR;
                cost.cardinality = (uint64_t)(child_cost.cardinality * SELECTIVITY_DEFAULT);
            }
            break;
        }
        
        case PLAN_PROJECT: {
            if (plan->data.project.child) {
                plan_cost_t child_cost = estimate_plan_cost(plan->data.project.child, ctx);
                cost = child_cost;
                cost.cpu_cost += child_cost.cardinality * CPU_COST_FACTOR * 0.1;
            }
            break;
        }
        
        case PLAN_JOIN: {
            if (plan->data.join.left && plan->data.join.right) {
                plan_cost_t left_cost = estimate_plan_cost(plan->data.join.left, ctx);
                plan_cost_t right_cost = estimate_plan_cost(plan->data.join.right, ctx);
                
                cost.io_cost = left_cost.io_cost + right_cost.io_cost;
                cost.cpu_cost = left_cost.cpu_cost + right_cost.cpu_cost + 
                               left_cost.cardinality * right_cost.cardinality * CPU_COST_FACTOR;
                cost.cardinality = (uint64_t)(left_cost.cardinality * right_cost.cardinality * SELECTIVITY_DEFAULT);
            }
            break;
        }
        
        case PLAN_SORT: {
            if (plan->data.sort.child) {
                plan_cost_t child_cost = estimate_plan_cost(plan->data.sort.child, ctx);
                cost = child_cost;
                if (child_cost.cardinality > 0) {
                    cost.cpu_cost += child_cost.cardinality * log2(child_cost.cardinality) * CPU_COST_FACTOR;
                }
            }
            break;
        }
        
        case PLAN_LIMIT: {
            if (plan->data.limit.child) {
                plan_cost_t child_cost = estimate_plan_cost(plan->data.limit.child, ctx);
                cost = child_cost;
                cost.cardinality = (plan->data.limit.limit < child_cost.cardinality) ? 
                                 plan->data.limit.limit : child_cost.cardinality;
            }
            break;
        }
        
        default:
            break;
    }
    
    cost.total_cost = cost.cpu_cost + cost.io_cost + cost.memory_cost;
    plan->cost = cost;
    
    return cost;
}

double estimate_selectivity(expression_t* condition, plan_stats_t* stats) {
    if (!condition) return 1.0;
    
    // Simplified selectivity estimation
    switch (condition->type) {
        case EXPR_BINARY_OP:
            switch (condition->data.binary_op.operator) {
                case TOKEN_EQUAL:
                    return stats ? (1.0 / stats->distinct_values) : 0.1;
                case TOKEN_LESS_THAN:
                case TOKEN_GREATER_THAN:
                    return 0.33;
                case TOKEN_LESS_EQUAL:
                case TOKEN_GREATER_EQUAL:
                    return 0.33;
                case TOKEN_AND:
                    return estimate_selectivity(condition->data.binary_op.left, stats) *
                           estimate_selectivity(condition->data.binary_op.right, stats);
                case TOKEN_OR:
                    return 1.0 - (1.0 - estimate_selectivity(condition->data.binary_op.left, stats)) *
                                 (1.0 - estimate_selectivity(condition->data.binary_op.right, stats));
                default:
                    return SELECTIVITY_DEFAULT;
            }
            break;
        default:
            return SELECTIVITY_DEFAULT;
    }
}

uint64_t estimate_cardinality(plan_node_t* plan, optimizer_context_t* ctx) {
    plan_cost_t cost = estimate_plan_cost(plan, ctx);
    return cost.cardinality;
}

// Cascades optimization algorithm implementation
int cascades_optimize_group(optimizer_context_t* ctx, memo_group_t* group) {
    if (!ctx || !group || group->optimized) return 0;
    
    // First, explore the group to generate all possible transformations
    if (cascades_explore_group(ctx, group) != 0) {
        return -1;
    }
    
    // Then optimize all expressions in the group
    plan_node_t* expr = group->expressions;
    while (expr) {
        // Use enhanced cost estimation with real statistics
        plan_cost_t cost = estimate_plan_cost_with_stats(expr, ctx);
        
        if (cost.total_cost < group->best_cost.total_cost) {
            group->best_cost = cost;
            group->best_plan = expr;
        }
        
        // Recursively optimize child groups
        for (uint32_t i = 0; i < expr->child_count; i++) {
            if (expr->children[i] && expr->children[i]->group) {
                cascades_optimize_group(ctx, expr->children[i]->group);
            }
        }
        
        expr = expr->next;
    }
    
    group->optimized = true;
    return 0;
}

int cascades_explore_group(optimizer_context_t* ctx, memo_group_t* group) {
    if (!ctx || !group || group->explored) return 0;
    
    // Apply transformation rules to generate new expressions
    cascades_apply_rules(ctx, group);
    
    group->explored = true;
    ctx->groups_explored++;
    
    return 0;
}

int cascades_apply_rules(optimizer_context_t* ctx, memo_group_t* group) {
    if (!ctx || !group) return -1;
    
    plan_node_t* expr = group->expressions;
    while (expr) {
        // Apply transformation rules
        for (uint32_t i = 0; i < ctx->transformation_rule_count; i++) {
            rule_t* rule = &ctx->transformation_rules[i];
            if (rule->condition && rule->condition(expr)) {
                plan_node_t* new_expr = rule->transform(expr, ctx);
                if (new_expr) {
                    memo_add_expression(ctx, group, new_expr);
                    ctx->rules_applied++;
                }
            }
        }
        
        // Apply implementation rules
        for (uint32_t i = 0; i < ctx->implementation_rule_count; i++) {
            rule_t* rule = &ctx->implementation_rules[i];
            if (rule->condition && rule->condition(expr)) {
                plan_node_t* new_expr = rule->transform(expr, ctx);
                if (new_expr) {
                    memo_add_expression(ctx, group, new_expr);
                    ctx->rules_applied++;
                }
            }
        }
        
        expr = expr->next;
    }
    
    return 0;
}

// Built-in transformation rules (simplified implementations)
bool rule_predicate_pushdown_condition(plan_node_t* node) {
    return node && node->type == PLAN_JOIN && 
           node->data.join.condition != NULL;
}

plan_node_t* rule_predicate_pushdown_transform(plan_node_t* node, optimizer_context_t* ctx) {
    // Simplified predicate pushdown: push filter down to left child
    if (!node || !ctx) return NULL;
    
    plan_node_t* new_node = plan_clone(node);
    if (new_node && new_node->data.join.left) {
        plan_node_t* filter = planner_create_filter_plan(new_node->data.join.condition, 
                                                        new_node->data.join.left);
        new_node->data.join.left = filter;
        new_node->data.join.condition = NULL;
    }
    
    return new_node;
}

bool rule_join_commutativity_condition(plan_node_t* node) {
    return node && node->type == PLAN_JOIN && 
           node->data.join.join_type == JOIN_INNER;
}

plan_node_t* rule_join_commutativity_transform(plan_node_t* node, optimizer_context_t* ctx) {
    if (!node || !ctx) return NULL;
    
    plan_node_t* new_node = plan_clone(node);
    if (new_node) {
        // Swap left and right children
        plan_node_t* temp = new_node->data.join.left;
        new_node->data.join.left = new_node->data.join.right;
        new_node->data.join.right = temp;
        
        // Update children array
        if (new_node->children && new_node->child_count >= 2) {
            temp = new_node->children[0];
            new_node->children[0] = new_node->children[1];
            new_node->children[1] = temp;
        }
    }
    
    return new_node;
}

// Additional placeholder rule implementations
bool rule_join_associativity_condition(plan_node_t* node) {
    return false; // Not implemented yet
}

plan_node_t* rule_join_associativity_transform(plan_node_t* node, optimizer_context_t* ctx) {
    return NULL; // Not implemented yet
}

bool rule_projection_pushdown_condition(plan_node_t* node) {
    return false; // Not implemented yet
}

plan_node_t* rule_projection_pushdown_transform(plan_node_t* node, optimizer_context_t* ctx) {
    return NULL; // Not implemented yet
}

bool rule_scan_to_index_scan_condition(plan_node_t* node) {
    return node && node->type == PLAN_SCAN && 
           node->data.scan.filter != NULL;
}

plan_node_t* rule_scan_to_index_scan_transform(plan_node_t* node, optimizer_context_t* ctx) {
    if (!node || !ctx) return NULL;
    
    // Create index scan plan
    return planner_create_index_scan_plan(node->data.scan.table_name, "idx_default", 
                                         node->data.scan.filter);
}

bool rule_join_to_nested_loop_condition(plan_node_t* node) {
    return node && node->type == PLAN_JOIN;
}

plan_node_t* rule_join_to_nested_loop_transform(plan_node_t* node, optimizer_context_t* ctx) {
    // For now, return the same node (implementation placeholder)
    return plan_clone(node);
}

bool rule_join_to_hash_join_condition(plan_node_t* node) {
    return node && node->type == PLAN_JOIN;
}

plan_node_t* rule_join_to_hash_join_transform(plan_node_t* node, optimizer_context_t* ctx) {
    // For now, return the same node (implementation placeholder)
    return plan_clone(node);
}

// Utility functions
void plan_print_tree(plan_node_t* plan, int indent) {
    if (!plan) return;
    
    for (int i = 0; i < indent; i++) {
        printf("  ");
    }
    
    printf("%s (cost=%.2f, card=%" PRIu64 ")\n", 
           plan_node_type_to_string(plan->type),
           plan->cost.total_cost,
           plan->cost.cardinality);
    
    for (uint32_t i = 0; i < plan->child_count; i++) {
        if (plan->children[i]) {
            plan_print_tree(plan->children[i], indent + 1);
        }
    }
}

void plan_destroy(plan_node_t* plan) {
    if (!plan) return;
    
    // Free node-specific data
    switch (plan->type) {
        case PLAN_SCAN:
            if (plan->data.scan.table_name) free(plan->data.scan.table_name);
            if (plan->data.scan.alias) free(plan->data.scan.alias);
            break;
            
        case PLAN_INDEX_SCAN:
            if (plan->data.index_scan.table_name) free(plan->data.index_scan.table_name);
            if (plan->data.index_scan.index_name) free(plan->data.index_scan.index_name);
            break;
            
        case PLAN_PROJECT:
            if (plan->data.project.expressions) free(plan->data.project.expressions);
            if (plan->data.project.column_names) {
                for (uint32_t i = 0; i < plan->data.project.expr_count; i++) {
                    if (plan->data.project.column_names[i]) {
                        free(plan->data.project.column_names[i]);
                    }
                }
                free(plan->data.project.column_names);
            }
            break;
            
        case PLAN_SORT:
            if (plan->data.sort.sort_keys) free(plan->data.sort.sort_keys);
            if (plan->data.sort.ascending) free(plan->data.sort.ascending);
            break;
            
        case PLAN_INSERT:
            if (plan->data.insert.table_name) free(plan->data.insert.table_name);
            if (plan->data.insert.column_names) {
                for (uint32_t i = 0; i < plan->data.insert.column_count; i++) {
                    if (plan->data.insert.column_names[i]) {
                        free(plan->data.insert.column_names[i]);
                    }
                }
                free(plan->data.insert.column_names);
            }
            if (plan->data.insert.values) free(plan->data.insert.values);
            break;
            
        case PLAN_UPDATE:
            if (plan->data.update.table_name) free(plan->data.update.table_name);
            if (plan->data.update.assignments) free(plan->data.update.assignments);
            break;
            
        case PLAN_DELETE:
            if (plan->data.delete.table_name) free(plan->data.delete.table_name);
            break;
            
        default:
            break;
    }
    
    // Free children
    if (plan->children) {
        for (uint32_t i = 0; i < plan->child_count; i++) {
            plan_destroy(plan->children[i]);
        }
        free(plan->children);
    }
    
    free(plan);
}

plan_node_t* plan_clone(plan_node_t* plan) {
    if (!plan) return NULL;
    
    // This is a simplified clone implementation
    // In a real implementation, we would need to deep-copy all fields
    plan_node_t* clone = create_plan_node(plan->type);
    if (!clone) return NULL;
    
    clone->cost = plan->cost;
    clone->optimization_level = plan->optimization_level;
    
    // Copy node-specific data based on type
    switch (plan->type) {
        case PLAN_SCAN: {
            clone->data.scan.table_name = strdup(plan->data.scan.table_name);
            clone->data.scan.alias = plan->data.scan.alias ? strdup(plan->data.scan.alias) : NULL;
            clone->data.scan.filter = plan->data.scan.filter ? plan_clone(plan->data.scan.filter) : NULL;
            clone->data.scan.stats = plan->data.scan.stats;
            break;
        }
        
        case PLAN_INDEX_SCAN: {
            clone->data.index_scan.table_name = strdup(plan->data.index_scan.table_name);
            clone->data.index_scan.index_name = strdup(plan->data.index_scan.index_name);
            clone->data.index_scan.index_condition = plan->data.index_scan.index_condition ? plan_clone(plan->data.index_scan.index_condition) : NULL;
            clone->data.index_scan.filter = plan->data.index_scan.filter ? plan_clone(plan->data.index_scan.filter) : NULL;
            clone->data.index_scan.is_unique = plan->data.index_scan.is_unique;
            clone->data.index_scan.stats = plan->data.index_scan.stats;
            break;
        }
        
        case PLAN_FILTER: {
            clone->data.filter.condition = plan->data.filter.condition ? plan_clone(plan->data.filter.condition) : NULL;
            clone->data.filter.child = plan->data.filter.child ? plan_clone(plan->data.filter.child) : NULL;
            break;
        }
        
        case PLAN_PROJECT: {
            clone->data.project.expr_count = plan->data.project.expr_count;
            clone->data.project.expressions = malloc(clone->data.project.expr_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.project.expr_count; i++) {
                clone->data.project.expressions[i] = plan->data.project.expressions[i] ? plan_clone(plan->data.project.expressions[i]) : NULL;
            }
            
            clone->data.project.column_names = malloc(clone->data.project.expr_count * sizeof(char*));
            for (uint32_t i = 0; i < clone->data.project.expr_count; i++) {
                clone->data.project.column_names[i] = plan->data.project.column_names[i] ? strdup(plan->data.project.column_names[i]) : NULL;
            }
            
            clone->data.project.child = plan->data.project.child ? plan_clone(plan->data.project.child) : NULL;
            break;
        }
        
        case PLAN_JOIN: {
            clone->data.join.join_type = plan->data.join.join_type;
            clone->data.join.condition = plan->data.join.condition ? plan_clone(plan->data.join.condition) : NULL;
            clone->data.join.left = plan->data.join.left ? plan_clone(plan->data.join.left) : NULL;
            clone->data.join.right = plan->data.join.right ? plan_clone(plan->data.join.right) : NULL;
            clone->data.join.stats = plan->data.join.stats;
            break;
        }
        
        case PLAN_AGGREGATE: {
            clone->data.aggregate.group_count = plan->data.aggregate.group_count;
            clone->data.aggregate.group_by = malloc(clone->data.aggregate.group_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.aggregate.group_count; i++) {
                clone->data.aggregate.group_by[i] = plan->data.aggregate.group_by[i] ? plan_clone(plan->data.aggregate.group_by[i]) : NULL;
            }
            
            clone->data.aggregate.agg_count = plan->data.aggregate.agg_count;
            clone->data.aggregate.aggregates = malloc(clone->data.aggregate.agg_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.aggregate.agg_count; i++) {
                clone->data.aggregate.aggregates[i] = plan->data.aggregate.aggregates[i] ? plan_clone(plan->data.aggregate.aggregates[i]) : NULL;
            }
            
            clone->data.aggregate.child = plan->data.aggregate.child ? plan_clone(plan->data.aggregate.child) : NULL;
            break;
        }
        
        case PLAN_SORT: {
            clone->data.sort.key_count = plan->data.sort.key_count;
            clone->data.sort.sort_keys = malloc(clone->data.sort.key_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.sort.key_count; i++) {
                clone->data.sort.sort_keys[i] = plan->data.sort.sort_keys[i] ? plan_clone(plan->data.sort.sort_keys[i]) : NULL;
            }
            
            clone->data.sort.ascending = malloc(clone->data.sort.key_count * sizeof(bool));
            for (uint32_t i = 0; i < clone->data.sort.key_count; i++) {
                clone->data.sort.ascending[i] = plan->data.sort.ascending[i];
            }
            
            clone->data.sort.child = plan->data.sort.child ? plan_clone(plan->data.sort.child) : NULL;
            break;
        }
        
        case PLAN_LIMIT: {
            clone->data.limit.limit = plan->data.limit.limit;
            clone->data.limit.offset = plan->data.limit.offset;
            clone->data.limit.child = plan->data.limit.child ? plan_clone(plan->data.limit.child) : NULL;
            break;
        }
        
        case PLAN_INSERT: {
            clone->data.insert.table_name = strdup(plan->data.insert.table_name);
            clone->data.insert.column_count = plan->data.insert.column_count;
            clone->data.insert.column_names = malloc(clone->data.insert.column_count * sizeof(char*));
            for (uint32_t i = 0; i < clone->data.insert.column_count; i++) {
                clone->data.insert.column_names[i] = plan->data.insert.column_names[i] ? strdup(plan->data.insert.column_names[i]) : NULL;
            }
            
            clone->data.insert.value_count = plan->data.insert.value_count;
            clone->data.insert.values = malloc(clone->data.insert.value_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.insert.value_count; i++) {
                clone->data.insert.values[i] = plan->data.insert.values[i] ? plan_clone(plan->data.insert.values[i]) : NULL;
            }
            break;
        }
        
        case PLAN_UPDATE: {
            clone->data.update.table_name = strdup(plan->data.update.table_name);
            clone->data.update.assignment_count = plan->data.update.assignment_count;
            clone->data.update.assignments = malloc(clone->data.update.assignment_count * sizeof(expression_t*));
            for (uint32_t i = 0; i < clone->data.update.assignment_count; i++) {
                clone->data.update.assignments[i] = plan->data.update.assignments[i] ? plan_clone(plan->data.update.assignments[i]) : NULL;
            }
            clone->data.update.condition = plan->data.update.condition ? plan_clone(plan->data.update.condition) : NULL;
            break;
        }
        
        case PLAN_DELETE: {
            clone->data.delete.table_name = strdup(plan->data.delete.table_name);
            clone->data.delete.condition = plan->data.delete.condition ? plan_clone(plan->data.delete.condition) : NULL;
            break;
        }
        
        default:
            break;
    }
    
    // Copy children
    clone->child_count = plan->child_count;
    if (clone->child_count > 0) {
        clone->children = malloc(clone->child_count * sizeof(plan_node_t*));
        for (uint32_t i = 0; i < clone->child_count; i++) {
            clone->children[i] = plan->children[i] ? plan_clone(plan->children[i]) : NULL;
        }
    } else {
        clone->children = NULL;
    }
    
    return clone;
}

bool plan_equals(plan_node_t* a, plan_node_t* b) {
    if (!a && !b) return true;
    if (!a || !b) return false;
    
    if (a->type != b->type) return false;
    if (a->child_count != b->child_count) return false;
    
    // Compare children recursively
    for (uint32_t i = 0; i < a->child_count; i++) {
        if (!plan_equals(a->children[i], b->children[i])) {
            return false;
        }
    }
    
    // Compare node-specific data based on type
    switch (a->type) {
        case PLAN_SCAN: {
            if (strcmp(a->data.scan.table_name, b->data.scan.table_name) != 0) return false;
            if ((a->data.scan.alias == NULL) != (b->data.scan.alias == NULL)) return false;
            if (a->data.scan.alias && strcmp(a->data.scan.alias, b->data.scan.alias) != 0) return false;
            if (!plan_equals(a->data.scan.filter, b->data.scan.filter)) return false;
            break;
        }
        
        case PLAN_INDEX_SCAN: {
            if (strcmp(a->data.index_scan.table_name, b->data.index_scan.table_name) != 0) return false;
            if (strcmp(a->data.index_scan.index_name, b->data.index_scan.index_name) != 0) return false;
            if (!plan_equals(a->data.index_scan.index_condition, b->data.index_scan.index_condition)) return false;
            if (!plan_equals(a->data.index_scan.filter, b->data.index_scan.filter)) return false;
            if (a->data.index_scan.is_unique != b->data.index_scan.is_unique) return false;
            break;
        }
        
        case PLAN_FILTER: {
            if (!plan_equals(a->data.filter.condition, b->data.filter.condition)) return false;
            if (!plan_equals(a->data.filter.child, b->data.filter.child)) return false;
            break;
        }
        
        case PLAN_PROJECT: {
            if (a->data.project.expr_count != b->data.project.expr_count) return false;
            for (uint32_t i = 0; i < a->data.project.expr_count; i++) {
                if (!plan_equals(a->data.project.expressions[i], b->data.project.expressions[i])) return false;
            }
            
            for (uint32_t i = 0; i < a->data.project.expr_count; i++) {
                if ((a->data.project.column_names[i] == NULL) != (b->data.project.column_names[i] == NULL)) return false;
                if (a->data.project.column_names[i] && strcmp(a->data.project.column_names[i], b->data.project.column_names[i]) != 0) return false;
            }
            if (!plan_equals(a->data.project.child, b->data.project.child)) return false;
            break;
        }
        
        case PLAN_JOIN: {
            if (a->data.join.join_type != b->data.join.join_type) return false;
            if (!plan_equals(a->data.join.condition, b->data.join.condition)) return false;
            if (!plan_equals(a->data.join.left, b->data.join.left)) return false;
            if (!plan_equals(a->data.join.right, b->data.join.right)) return false;
            break;
        }
        
        case PLAN_AGGREGATE: {
            if (a->data.aggregate.group_count != b->data.aggregate.group_count) return false;
            for (uint32_t i = 0; i < a->data.aggregate.group_count; i++) {
                if (!plan_equals(a->data.aggregate.group_by[i], b->data.aggregate.group_by[i])) return false;
            }
            
            if (a->data.aggregate.agg_count != b->data.aggregate.agg_count) return false;
            for (uint32_t i = 0; i < a->data.aggregate.agg_count; i++) {
                if (!plan_equals(a->data.aggregate.aggregates[i], b->data.aggregate.aggregates[i])) return false;
            }
            if (!plan_equals(a->data.aggregate.child, b->data.aggregate.child)) return false;
            break;
        }
        
        case PLAN_SORT: {
            if (a->data.sort.key_count != b->data.sort.key_count) return false;
            for (uint32_t i = 0; i < a->data.sort.key_count; i++) {
                if (!plan_equals(a->data.sort.sort_keys[i], b->data.sort.sort_keys[i])) return false;
            }
            
            for (uint32_t i = 0; i < a->data.sort.key_count; i++) {
                if (a->data.sort.ascending[i] != b->data.sort.ascending[i]) return false;
            }
            if (!plan_equals(a->data.sort.child, b->data.sort.child)) return false;
            break;
        }
        
        case PLAN_LIMIT: {
            if (a->data.limit.limit != b->data.limit.limit) return false;
            if (a->data.limit.offset != b->data.limit.offset) return false;
            if (!plan_equals(a->data.limit.child, b->data.limit.child)) return false;
            break;
        }
        
        case PLAN_INSERT: {
            if (strcmp(a->data.insert.table_name, b->data.insert.table_name) != 0) return false;
            if (a->data.insert.column_count != b->data.insert.column_count) return false;
            for (uint32_t i = 0; i < a->data.insert.column_count; i++) {
                if ((a->data.insert.column_names[i] == NULL) != (b->data.insert.column_names[i] == NULL)) return false;
                if (a->data.insert.column_names[i] && strcmp(a->data.insert.column_names[i], b->data.insert.column_names[i]) != 0) return false;
            }
            
            if (a->data.insert.value_count != b->data.insert.value_count) return false;
            for (uint32_t i = 0; i < a->data.insert.value_count; i++) {
                if (!plan_equals(a->data.insert.values[i], b->data.insert.values[i])) return false;
            }
            break;
        }
        
        case PLAN_UPDATE: {
            if (strcmp(a->data.update.table_name, b->data.update.table_name) != 0) return false;
            if (a->data.update.assignment_count != b->data.update.assignment_count) return false;
            for (uint32_t i = 0; i < a->data.update.assignment_count; i++) {
                if (!plan_equals(a->data.update.assignments[i], b->data.update.assignments[i])) return false;
            }
            if (!plan_equals(a->data.update.condition, b->data.update.condition)) return false;
            break;
        }
        
        case PLAN_DELETE: {
            if (strcmp(a->data.delete.table_name, b->data.delete.table_name) != 0) return false;
            if (!plan_equals(a->data.delete.condition, b->data.delete.condition)) return false;
            break;
        }
        
        default:
            break;
    }
    
    return true;
}

// String conversion functions
const char* plan_node_type_to_string(plan_node_type_t type) {
    switch (type) {
        case PLAN_SCAN: return "Scan";
        case PLAN_INDEX_SCAN: return "IndexScan";
        case PLAN_FILTER: return "Filter";
        case PLAN_PROJECT: return "Project";
        case PLAN_JOIN: return "Join";
        case PLAN_AGGREGATE: return "Aggregate";
        case PLAN_SORT: return "Sort";
        case PLAN_LIMIT: return "Limit";
        case PLAN_INSERT: return "Insert";
        case PLAN_UPDATE: return "Update";
        case PLAN_DELETE: return "Delete";
        case PLAN_UNION: return "Union";
        case PLAN_SUBQUERY: return "Subquery";
        default: return "Unknown";
    }
}

const char* join_type_to_string(join_type_t type) {
    switch (type) {
        case JOIN_INNER: return "Inner";
        case JOIN_LEFT_OUTER: return "LeftOuter";
        case JOIN_RIGHT_OUTER: return "RightOuter";
        case JOIN_FULL_OUTER: return "FullOuter";
        case JOIN_CROSS: return "Cross";
        case JOIN_SEMI: return "Semi";
        case JOIN_ANTI: return "Anti";
        default: return "Unknown";
    }
}

// Initialize default rules
static void init_default_rules(optimizer_context_t* ctx) {
    if (!ctx) return;
    
    // Initialize transformation rules (simplified)
    ctx->transformation_rule_count = 2;
    ctx->transformation_rules = calloc(ctx->transformation_rule_count, sizeof(rule_t));
    
    if (ctx->transformation_rules) {
        ctx->transformation_rules[0].name = "PredicatePushdown";
        ctx->transformation_rules[0].condition = rule_predicate_pushdown_condition;
        ctx->transformation_rules[0].transform = rule_predicate_pushdown_transform;
        ctx->transformation_rules[0].priority = 1.0;
        ctx->transformation_rules[0].is_implementation = false;
        
        ctx->transformation_rules[1].name = "JoinCommutativity";
        ctx->transformation_rules[1].condition = rule_join_commutativity_condition;
        ctx->transformation_rules[1].transform = rule_join_commutativity_transform;
        ctx->transformation_rules[1].priority = 0.8;
        ctx->transformation_rules[1].is_implementation = false;
    }
    
    // Initialize implementation rules (simplified)
    ctx->implementation_rule_count = 1;
    ctx->implementation_rules = calloc(ctx->implementation_rule_count, sizeof(rule_t));
    
    if (ctx->implementation_rules) {
        ctx->implementation_rules[0].name = "ScanToIndexScan";
        ctx->implementation_rules[0].condition = rule_scan_to_index_scan_condition;
        ctx->implementation_rules[0].transform = rule_scan_to_index_scan_transform;
        ctx->implementation_rules[0].priority = 1.0;
        ctx->implementation_rules[0].is_implementation = true;
    }
}

// Additional placeholder functions for missing implementations
static plan_node_t* planner_create_insert_plan(query_planner_t* planner, statement_t* stmt) {
    if (!planner || !stmt || stmt->type != STMT_INSERT) return NULL;
    // Placeholder implementation
    return create_plan_node(PLAN_INSERT);
}

static plan_node_t* planner_create_update_plan(query_planner_t* planner, statement_t* stmt) {
    if (!planner || !stmt || stmt->type != STMT_UPDATE) return NULL;
    // Placeholder implementation
    return create_plan_node(PLAN_UPDATE);
}

static plan_node_t* planner_create_delete_plan(query_planner_t* planner, statement_t* stmt) {
    if (!planner || !stmt || stmt->type != STMT_DELETE) return NULL;
    // Placeholder implementation
    return create_plan_node(PLAN_DELETE);
}

// Configuration and debugging functions (placeholders)
int planner_set_optimization_level(query_planner_t* planner, uint32_t level) {
    if (!planner || level > 2) return -1;
    planner->optimization_level = level;
    return 0;
}

int planner_enable_rule(query_planner_t* planner, const char* rule_name, bool enabled) {
    // Placeholder implementation
    return 0;
}

// Helper function to count expressions in a plan node
static uint32_t count_expressions_in_plan(plan_node_t* plan) {
    if (!plan) return 0;
    
    uint32_t count = 0;
    
    switch (plan->type) {
        case PLAN_SCAN:
            if (plan->data.scan.filter) count++;
            break;
            
        case PLAN_INDEX_SCAN:
            if (plan->data.index_scan.index_condition) count++;
            if (plan->data.index_scan.filter) count++;
            break;
            
        case PLAN_FILTER:
            if (plan->data.filter.condition) count++;
            break;
            
        case PLAN_PROJECT:
            count += plan->data.project.expr_count;
            break;
            
        case PLAN_JOIN:
            if (plan->data.join.condition) count++;
            break;
            
        case PLAN_AGGREGATE:
            count += plan->data.aggregate.group_count;
            count += plan->data.aggregate.agg_count;
            break;
            
        case PLAN_SORT:
            count += plan->data.sort.key_count;
            break;
            
        case PLAN_INSERT:
            count += plan->data.insert.value_count;
            break;
            
        case PLAN_UPDATE:
            count += plan->data.update.assignment_count;
            if (plan->data.update.condition) count++;
            break;
            
        case PLAN_DELETE:
            if (plan->data.delete.condition) count++;
            break;
            
        default:
            break;
    }
    
    // Count expressions in children
    for (uint32_t i = 0; i < plan->child_count; i++) {
        count += count_expressions_in_plan(plan->children[i]);
    }
    
    return count;
}

static uint32_t count_expressions_in_group(memo_group_t* group) {
    if (!group) return 0;
    
    uint32_t count = 0;
    plan_node_t* plan = group->expressions;
    while (plan) {
        count += count_expressions_in_plan(plan);
        plan = plan->next;
    }
    
    return count;
}

void planner_print_memo(optimizer_context_t* ctx) {
    if (!ctx) return;
    
    printf("Memo Table (%" PRIu32 " groups):\n", ctx->memo_group_count);
    for (uint32_t i = 0; i < ctx->memo_table_size; i++) {
        memo_group_t* group = ctx->memo_table[i];
        while (group) {
            printf("  Group %" PRIu32 ": cost=%.2f, expressions=%d\n",
                   group->group_id, group->best_cost.total_cost,
                   count_expressions_in_group(group));
            group = group->next;
        }
    }
}

void planner_print_stats(query_planner_t* planner) {
    if (!planner) return;
    
    printf("Query Planner Statistics:\n");
    printf("  Plans generated: %" PRIu64 "\n", planner->plans_generated);
    printf("  Optimization time: %" PRIu64 " μs\n", planner->optimization_time_us);
    printf("  Optimization level: %" PRIu32 "\n", planner->optimization_level);
    
    if (planner->optimizer) {
        printf("  Rules applied: %" PRIu32 "\n", planner->optimizer->rules_applied);
        printf("  Groups explored: %" PRIu32 "\n", planner->optimizer->groups_explored);
        printf("  Plans considered: %" PRIu32 "\n", planner->optimizer->plans_considered);
    }
}

// Configuration integration functions
int optimizer_load_config_params(optimizer_context_t* ctx) {
    if (!ctx || !ctx->config_manager) {
        return -1;
    }
    
    config_manager_t* config = ctx->config_manager;
    
    // Load cost model parameters from configuration
    ctx->cpu_cost_factor = config_get_double(config, "cpu_cost_factor", CPU_COST_FACTOR);
    ctx->io_cost_factor = config_get_double(config, "io_cost_factor", IO_COST_FACTOR);
    ctx->memory_cost_factor = config_get_double(config, "memory_cost_factor", MEMORY_COST_FACTOR);
    ctx->index_selectivity_boost = config_get_double(config, "index_selectivity_boost", 0.5);
    
    // Load optimizer configuration
    ctx->max_iterations = config_get_uint(config, "max_optimization_iterations", DEFAULT_MAX_ITERATIONS);
    ctx->cost_threshold = config_get_double(config, "cost_threshold", DEFAULT_COST_THRESHOLD);
    ctx->hash_join_threshold = config_get_uint(config, "hash_join_threshold", 1000);
    
    // Load optimizer flags
    ctx->enable_join_reordering = config_get_bool(config, "enable_join_reordering", true);
    ctx->enable_predicate_pushdown = config_get_bool(config, "enable_predicate_pushdown", true);
    ctx->enable_index_selection = config_get_bool(config, "enable_index_selection", true);
    
    return 0;
}

int optimizer_update_cost_factors(optimizer_context_t* ctx) {
    if (!ctx) {
        return -1;
    }
    
    // This could be called when configuration changes
    return optimizer_load_config_params(ctx);
}

// Enhanced cost estimation with real statistics
plan_cost_t estimate_plan_cost_with_stats(plan_node_t* plan, optimizer_context_t* ctx) {
    plan_cost_t cost = {0};
    
    if (!plan || !ctx) return cost;
    
    switch (plan->type) {
        case PLAN_SCAN: {
            const char* table_name = plan->data.scan.table_name;
            if (table_name && ctx->stats_collector) {
                // Try to get real table statistics
                // For simplicity, assume table_id = hash of table name
                uint32_t table_id = stats_hash_query(ctx->stats_collector, table_name) % 1000;
                table_stats_t* stats = stats_get_table_stats(ctx->stats_collector, table_id);
                
                if (stats) {
                    uint64_t rows = stats->row_count;
                    cost.io_cost = stats->page_count * ctx->io_cost_factor;
                    cost.cpu_cost = rows * ctx->cpu_cost_factor;
                    cost.cardinality = rows;
                    
                    // Apply filter selectivity if present
                    if (plan->data.scan.filter) {
                        double selectivity = estimate_selectivity_with_stats(
                            plan->data.scan.filter, table_id, 0, ctx);
                        cost.cardinality = (uint64_t)(cost.cardinality * selectivity);
                    }
                } else {
                    // Fallback to original estimate
                    return estimate_plan_cost(plan, ctx);
                }
            } else {
                return estimate_plan_cost(plan, ctx);
            }
            break;
        }
        
        case PLAN_INDEX_SCAN: {
            const char* table_name = plan->data.index_scan.table_name;
            if (table_name && ctx->stats_collector) {
                uint32_t table_id = stats_hash_query(ctx->stats_collector, table_name) % 1000;
                table_stats_t* stats = stats_get_table_stats(ctx->stats_collector, table_id);
                
                if (stats) {
                    // Index scans are more selective
                    uint64_t rows = (uint64_t)(stats->row_count * ctx->index_selectivity_boost);
                    cost.io_cost = (stats->page_count / 10) * ctx->io_cost_factor;  // Fewer pages
                    cost.cpu_cost = rows * ctx->cpu_cost_factor * 0.5;  // More efficient CPU usage
                    cost.cardinality = rows;
                    
                    // Apply index condition selectivity
                    if (plan->data.index_scan.index_condition) {
                        double selectivity = estimate_selectivity_with_stats(
                            plan->data.index_scan.index_condition, table_id, 0, ctx);
                        cost.cardinality = (uint64_t)(cost.cardinality * selectivity);
                    }
                } else {
                    return estimate_plan_cost(plan, ctx);
                }
            } else {
                return estimate_plan_cost(plan, ctx);
            }
            break;
        }
        
        case PLAN_JOIN: {
            if (plan->data.join.left && plan->data.join.right) {
                plan_cost_t left_cost = estimate_plan_cost_with_stats(plan->data.join.left, ctx);
                plan_cost_t right_cost = estimate_plan_cost_with_stats(plan->data.join.right, ctx);
                
                cost.io_cost = left_cost.io_cost + right_cost.io_cost;
                cost.cpu_cost = left_cost.cpu_cost + right_cost.cpu_cost;
                
                // Choose join algorithm based on cardinality
                if (left_cost.cardinality > ctx->hash_join_threshold || 
                    right_cost.cardinality > ctx->hash_join_threshold) {
                    // Hash join cost
                    cost.cpu_cost += (left_cost.cardinality + right_cost.cardinality) * ctx->cpu_cost_factor;
                    cost.memory_cost = (left_cost.cardinality < right_cost.cardinality ? 
                                       left_cost.cardinality : right_cost.cardinality) * ctx->memory_cost_factor;
                } else {
                    // Nested loop join cost
                    cost.cpu_cost += left_cost.cardinality * right_cost.cardinality * ctx->cpu_cost_factor;
                }
                
                // Estimate join result cardinality
                cost.cardinality = estimate_join_cardinality(
                    plan->data.join.left, plan->data.join.right, 
                    plan->data.join.condition, ctx);
            }
            break;
        }
        
        default:
            // For other plan types, fall back to original estimation
            return estimate_plan_cost(plan, ctx);
    }
    
    cost.total_cost = cost.cpu_cost + cost.io_cost + cost.memory_cost;
    plan->cost = cost;
    
    return cost;
}

double estimate_selectivity_with_stats(expression_t* condition, uint32_t table_id, 
                                      uint32_t column_id, optimizer_context_t* ctx) {
    if (!condition || !ctx || !ctx->stats_collector) {
        return SELECTIVITY_DEFAULT;
    }
    
    // Use the statistics collector to get more accurate selectivity
    switch (condition->type) {
        case EXPR_BINARY_OP: {
            switch (condition->data.binary_op.operator) {
                case TOKEN_EQUAL: {
                    // For equality conditions, use column statistics
                    column_stats_t* col_stats = stats_get_column_stats(
                        ctx->stats_collector, table_id, column_id);
                    if (col_stats && col_stats->distinct_count > 0) {
                        return 1.0 / col_stats->distinct_count;
                    }
                    return 0.1;  // Default for equality
                }
                
                case TOKEN_LESS_THAN:
                case TOKEN_GREATER_THAN:
                case TOKEN_LESS_EQUAL:
                case TOKEN_GREATER_EQUAL: {
                    // For range conditions, could use histogram if available
                    return 0.33;  // Default for range queries
                }
                
                case TOKEN_AND: {
                    double left_sel = estimate_selectivity_with_stats(
                        condition->data.binary_op.left, table_id, column_id, ctx);
                    double right_sel = estimate_selectivity_with_stats(
                        condition->data.binary_op.right, table_id, column_id, ctx);
                    return left_sel * right_sel;  // Independence assumption
                }
                
                case TOKEN_OR: {
                    double left_sel = estimate_selectivity_with_stats(
                        condition->data.binary_op.left, table_id, column_id, ctx);
                    double right_sel = estimate_selectivity_with_stats(
                        condition->data.binary_op.right, table_id, column_id, ctx);
                    return 1.0 - (1.0 - left_sel) * (1.0 - right_sel);
                }
                
                default:
                    return SELECTIVITY_DEFAULT;
            }
        }
        
        default:
            return SELECTIVITY_DEFAULT;
    }
}

uint64_t estimate_join_cardinality(plan_node_t* left, plan_node_t* right, 
                                  expression_t* condition, optimizer_context_t* ctx) {
    if (!left || !right || !ctx) {
        return 1000;  // Default estimate
    }
    
    plan_cost_t left_cost = estimate_plan_cost_with_stats(left, ctx);
    plan_cost_t right_cost = estimate_plan_cost_with_stats(right, ctx);
    
    if (!condition) {
        // Cross join - Cartesian product
        return left_cost.cardinality * right_cost.cardinality;
    }
    
    // For equi-joins, use simple heuristic
    // In practice, this should consider foreign key relationships,
    // value distributions, etc.
    double join_selectivity = 0.1;  // Default join selectivity
    
    return (uint64_t)(left_cost.cardinality * right_cost.cardinality * join_selectivity);
}